package com.chukun.flink.stream.action.click;

import org.apache.flink.api.common.eventtime.Watermark;
import org.apache.flink.api.common.eventtime.WatermarkGenerator;
import org.apache.flink.api.common.eventtime.WatermarkOutput;

import java.io.Serializable;

/**
     * watermarker生成策略
     */
public class PeriodicWatermarkGenerator implements WatermarkGenerator<UserBehaviorEvent>, Serializable {

    private long currentWatermark = Long.MIN_VALUE;

    @Override
    public void onEvent(UserBehaviorEvent event, long eventTimestamp, WatermarkOutput output) {
        currentWatermark = eventTimestamp;
    }

    @Override
    public void onPeriodicEmit(WatermarkOutput output) {
        long effectiveWatermark =
                    currentWatermark == Long.MIN_VALUE ? Long.MIN_VALUE : currentWatermark - 1;
        output.emitWatermark(new Watermark(effectiveWatermark));
    }
}